#include "rpc_channel_server.h"

namespace xlcomm {
namespace net {

RpcChannelServer::RpcChannelServer(net::EventLoop* loop,
                                   const net::InetAddress& listen_addr)
    : loop_(loop), server_(loop, listen_addr, "RpcChannelServer") {
  codec_.set_rpc_message_cb(
      std::bind(&RpcChannelServer::OnRpcMessage, this, _1, _2));
  server_.set_message_callback(
      std::bind(&RpcMessageCodec::OnMessage, codec_, _1, _2, _3));
}

void RpcChannelServer::RegisterService(google::protobuf::Service* service) {
  const google::protobuf::ServiceDescriptor* desc = service->GetDescriptor();
  services_[desc->full_name()] = service;
  LOG_DEBUG << "RpcChannelServer::RegisterService==== desc->full_name():"
            << desc->full_name();
}

void RpcChannelServer::OnRpcMessage(const net::TcpConnectionPtr& conn,
                                    const RpcMessage& rpcmsg) {
  LOG_DEBUG << "RpcChannelServer::OnRpcMessage==== in, services_.size:"
            << services_.size() << " rpcmsg.service():" << rpcmsg.service();
  if (rpcmsg.type() != RPC_TYPE_REQUEST) return;

  ServiceMap::iterator it = services_.find(rpcmsg.service());
  if (it != services_.end()) {
    LOG_DEBUG << "RpcChannelServer::OnRpcMessage==== find";
    google::protobuf::Service* service = it->second;
    const google::protobuf::ServiceDescriptor* service_desc =
        service->GetDescriptor();
    const google::protobuf::MethodDescriptor* method_desc =
        service_desc->FindMethodByName(rpcmsg.method());

    if (method_desc) {
      LOG_DEBUG << "RpcChannelServer::OnRpcMessage==== method_desc";
      std::unique_ptr<google::protobuf::Message> request(
          service->GetRequestPrototype(method_desc).New());

      if (request->ParseFromString(rpcmsg.request())) {
        LOG_DEBUG << "RpcChannelServer::OnRpcMessage==== ParseFromString";
        std::unique_ptr<google::protobuf::Message> response(
            service->GetResponsePrototype(method_desc).New());
        LOG_DEBUG << "RpcChannelServer::OnRpcMessage==== service->CallMethod";
        service->CallMethod(method_desc, NULL, request.get(), response.get(),
                            NULL);
        LOG_DEBUG
            << "RpcChannelServer::OnRpcMessage==== service->CallMethod ok";

        RpcMessage retmsg;
        retmsg.set_type(RPC_TYPE_RESPONSE);
        retmsg.set_id(rpcmsg.id());
        retmsg.set_service(rpcmsg.service());
        retmsg.set_method(rpcmsg.method());
        std::string response_str;
        response->SerializeToString(&response_str);
        retmsg.set_response(response_str);
        retmsg.set_error(RPC_ERR_OK);
        codec_.SendRpcMessage(conn, retmsg);

        LOG_DEBUG << "RpcChannelServer::OnRpcMessage==== retmsg.method:"
                  << rpcmsg.method() << " Send ok";
      }
    }
  }
}

}  // namespace net
}  // namespace xlcomm